package top.pcat.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import top.pcat.config.KafkaConfig;
import top.pcat.config.ServerConfig;
import top.pcat.service.MessagingService;

import java.time.Duration;
import java.util.*;

@Slf4j
public class KafkaProducerUtil {

    private static volatile KafkaProducer<String, byte[]> producer; // Kafka 生产者

    private static AdminClient kafkaAdminClient;

    private KafkaProducerUtil() {}

    public KafkaProducerUtil(KafkaConfig kafkaConfig) {
        if (producer == null) {
            synchronized (KafkaProducerUtil.class) {
                if (producer == null) {
                    Properties props = new Properties();
                    props.put("bootstrap.servers", kafkaConfig.getBootstrapServers());
                    // 配置 Kafka 生产者
                    props.put("key.serializer", kafkaConfig.getProducerKeySerializer());
                    props.put("value.serializer", kafkaConfig.getProducerValueSerializer());
                    producer = new KafkaProducer<>(props);
                    kafkaAdminClient = AdminClient.create(props);
                }
            }
        }

    }

    /**
     * 新增话题
     *
     * @param topicName 话题名称
     */
    public void newTopic(String topicName) {
        NewTopic newTopic = new NewTopic(topicName, 3, (short) 1);
        kafkaAdminClient.createTopics(Collections.singletonList(newTopic));
    }

    /**
     * 删除Topic
     *
     * @param topicName 话题名称
     */
    public void delTopic(String topicName) {
        kafkaAdminClient.deleteTopics(Collections.singletonList(topicName));
    }

    /**
     * 发送消息到 Kafka 话题
     *
     * @param topic 话题名称
     * @param bytes 数据
     */
    public void sendMsgByKafka(String topic, byte[] bytes) {
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, ServerConfig.serverId, bytes);
        log.info("发送kafka消息 ---->  key = " + record.key() +" toString = "+record.toString()+" value = "+record.value());
        producer.send(record);
    }


}